package com.asura.ops.sync.check.server.service.impl;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.spring.boot.autoconfigure.DruidDataSourceBuilder;
import com.asura.ops.sync.check.server.dto.CfgDbDto;
import com.asura.ops.sync.check.server.entity.CfgCheckPointEntity;
import com.asura.ops.sync.check.server.entity.CfgClientDbEntity;
import com.asura.ops.sync.check.server.entity.CfgClientTableFieldEntity;
import com.asura.ops.sync.check.server.mapper.CfgCheckPointMapper;
import com.asura.ops.sync.check.server.mapper.CfgDbMapper;
import com.asura.ops.sync.check.server.req.CheckDataReq;
import com.asura.ops.sync.check.server.service.ManageService;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author: huyuntao(Mars)
 * @date: 2022/8/10
 * @description: 类的描述
 */
@Service
@Slf4j
public class ManageServiceImpl implements ManageService {

    //源连接池
    private final Map<Long, JdbcTemplate> sourceDataSourceMap = Maps.newHashMap();

    //客户端连接池
    private final Map<Long, JdbcTemplate> clientDataSourceMap = Maps.newHashMap();

    @Resource
    private CfgDbMapper dbMapper;

    @Resource
    private CfgCheckPointMapper checkPointMapper;

    private final List<String> ignoreFieldList = Lists.newArrayList("rowversion", "updated_at", "updated_name", "updated_by", "created_at", "created_name", "created_by", "last_updated_at");

    private final List<String> specialUpdatedTimeFiledTableList = Lists.newArrayList("tt_sign_report_package", "tt_provider_service_mode", "sign_report_provider_detail", "sign_report_provider");

    private final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";

    private final Integer betweenHour = 2;

    private Boolean isChecking = false;

    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);


    @PostConstruct
    public void init() {
        scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                autoCheck();
            }
        }, 0, 1, TimeUnit.HOURS);
    }

    @Override
    public void checkData(CheckDataReq checkDataReq) throws Exception {
        log.info("【执行check】:{}", JSONUtil.toJsonStr(checkDataReq));

        CfgDbDto cfgDb = dbMapper.selectCfgDbBy(checkDataReq.getClientCode(), checkDataReq.getClientTable());
        if (cfgDb == null) {
            log.info("获取db连接为空");
            throw new Exception("获取db连接为空");
        }

        //获取源端数量
        JdbcTemplate sourceDataSource = buildSourceDataSource(cfgDb);
        Integer sourceDataCount = queryDataCount(checkDataReq, cfgDb.getTableName(), sourceDataSource);

        CfgClientDbEntity cfgClientDb = dbMapper.selectCfgClientDbBy(checkDataReq.getClientCode(), checkDataReq.getClientDbName());
        if (cfgDb == null) {
            log.info("获取Client db连接为空");
            throw new Exception("获取Client db连接为空");
        }

        //获取client端数量
        JdbcTemplate clientDataSource = buildClientDataSource(cfgClientDb);
        Integer clientDataCount = queryDataCount(checkDataReq, checkDataReq.getClientTable(), clientDataSource);

        if (sourceDataCount.intValue() != clientDataCount.intValue()) {
            String msg = String.format("%s,%s,%s {%s}至{%s}数据量不一致", checkDataReq.getClientCode(), checkDataReq.getClientDbName(), checkDataReq.getClientTable(), sourceDataCount, clientDataCount);
            log.info(msg);
            throw new Exception(msg);
        }
        //获取client端关注的字段，只对比关注的字段
        List<CfgClientTableFieldEntity> fields = dbMapper.selectCfgClientTableFieldBy(checkDataReq.getClientCode(), checkDataReq.getClientDbName(), checkDataReq.getClientTable());
        if (CollectionUtils.isEmpty(fields)) {
            String msg = String.format("没有设置关注字段:%s,%s,%s", checkDataReq.getClientCode(), checkDataReq.getClientDbName(), checkDataReq.getClientTable());
            log.info(msg);
            throw new Exception(msg);
        }

        Integer pageSize = 10000;
        Integer totalNum = clientDataCount / pageSize + (clientDataCount % pageSize == 0 ? 0 : 1);

        for (Integer pageNum = 1; pageNum <= totalNum; pageNum++) {
            Map<Long, Map<String, Object>> sourceData = queryData(checkDataReq, cfgDb.getTableName(), sourceDataSource, pageSize, pageNum);
            Map<Long, Map<String, Object>> clientData = queryData(checkDataReq, checkDataReq.getClientTable(), clientDataSource, pageSize, pageNum);
            if (sourceData.size() != sourceData.size()) {
                String msg = String.format("查询数据:{}至{}数据量不一致", checkDataReq.getStartTime(), checkDataReq.getEndTime());
                log.info(msg);
                throw new Exception(msg);
            }

            List<String> focusFieldList = fields.stream().map(f -> f.getClientTableField()).collect(Collectors.toList());
            compareData(sourceData, clientData, checkDataReq, focusFieldList);
        }

        if (checkDataReq.getIsAuto()) {
            saveCheckPoint(checkDataReq);
        }
    }

    public void autoCheck() {
        if (isChecking) {
            log.info("有任务正在checking");
            return;
        }
        isChecking = true;

        try {
            List<CfgClientDbEntity> allClientDb = dbMapper.selectCfgClientDbList();
            if (CollectionUtils.isEmpty(allClientDb)) {
                log.info("没有配置客户端书库连接,无需对比数据");
                return;
            }

            for (CfgClientDbEntity clientDb : allClientDb) {
                List<CfgClientTableFieldEntity> tables = dbMapper.selectCfgClientTable(clientDb.getClientCode(), clientDb.getClientDbName());
                if (CollectionUtils.isEmpty(tables)) {
                    log.info("{},{}未配置待对比检查的表字段,无需对比数据", clientDb.getClientCode(), clientDb.getClientDbName());
                    continue;
                }

                for (CfgClientTableFieldEntity table : tables) {
                    CheckDataReq checkData = new CheckDataReq();
                    checkData.setClientCode(table.getClientCode());
                    checkData.setClientDbName(table.getClientDbName());
                    checkData.setClientTable(table.getClientTable());
                    if (specialUpdatedTimeFiledTableList.contains(table.getClientTable())) {
                        checkData.setUpdateTimeFieldName("last_updated_at");
                    } else {
                        checkData.setUpdateTimeFieldName("updated_at");
                    }

                    CfgCheckPointEntity checkPoint = getCheckPoint(checkData);
                    if (checkPoint == null) {
                        log.info("{},{},{}未配置检查点,创建一个当前时间的检查点", table.getClientCode(), table.getClientDbName(), table.getClientTable());
                        checkPoint = saveCheckPoint(table.getClientCode(), table.getClientDbName(), table.getClientTable());
                    }

                    LocalDateTime checkPointDateTime = LocalDateTime.parse(checkPoint.getCheckTimePoint(), DateTimeFormatter.ofPattern(DATE_FORMAT));
                    while (true) {
                        LocalDateTime endDateTime = checkPointDateTime.plusHours(betweenHour);
                        if (endDateTime.isBefore(LocalDateTime.now())) {
                            checkData.setIsAuto(true);
                            checkData.setStartTime(checkPointDateTime.format(DateTimeFormatter.ofPattern(DATE_FORMAT)));
                            checkData.setEndTime(endDateTime.format(DateTimeFormatter.ofPattern(DATE_FORMAT)));
                            //执行check
                            try {
                                checkData(checkData);
                            } catch (Exception ex) {
                                log.error("执行checkData异常:{}", ex);
                                break;
                            }
                            checkPointDateTime = endDateTime;
                        } else {
                            break;
                        }
                    }
                }
            }
        } catch (Exception ex) {
            log.error("check异常:{}", ex);
        } finally {
            isChecking = false;
        }
    }

    /**
     * 保存检查点
     *
     * @param checkDataReq
     */
    private void saveCheckPoint(CheckDataReq checkDataReq) {
        LambdaQueryWrapper<CfgCheckPointEntity> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(CfgCheckPointEntity::getClientCode, checkDataReq.getClientCode());
        queryWrapper.eq(CfgCheckPointEntity::getClientDbName, checkDataReq.getClientDbName());
        queryWrapper.eq(CfgCheckPointEntity::getClientTable, checkDataReq.getClientTable());

        CfgCheckPointEntity checkPoint = checkPointMapper.selectOne(queryWrapper);
        if (checkPoint == null) {
            checkPoint = new CfgCheckPointEntity();
        }

        checkPoint.setClientCode(checkDataReq.getClientCode());
        checkPoint.setClientDbName(checkDataReq.getClientDbName());
        checkPoint.setClientTable(checkDataReq.getClientTable());
        checkPoint.setCheckTimePoint(checkDataReq.getEndTime());

        if (checkPoint.getId() == null) {
            checkPointMapper.insert(checkPoint);
        } else {
            checkPointMapper.updateById(checkPoint);
        }
    }

    private CfgCheckPointEntity saveCheckPoint(String clientCode, String clientDbName, String clientTable) {
        CfgCheckPointEntity checkPoint = checkPoint = new CfgCheckPointEntity();
        checkPoint.setClientCode(clientCode);
        checkPoint.setClientDbName(clientDbName);
        checkPoint.setClientTable(clientTable);
        checkPoint.setCheckTimePoint(DateUtil.format(new Date(), DATE_FORMAT));

        checkPointMapper.insert(checkPoint);

        return checkPoint;
    }

    /**
     * 获取检查点
     *
     * @param checkDataReq
     * @return
     */
    private CfgCheckPointEntity getCheckPoint(CheckDataReq checkDataReq) {
        LambdaQueryWrapper<CfgCheckPointEntity> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(CfgCheckPointEntity::getClientCode, checkDataReq.getClientCode());
        queryWrapper.eq(CfgCheckPointEntity::getClientDbName, checkDataReq.getClientDbName());
        queryWrapper.eq(CfgCheckPointEntity::getClientTable, checkDataReq.getClientTable());

        CfgCheckPointEntity checkPoint = checkPointMapper.selectOne(queryWrapper);
        return checkPoint;
    }

    /**
     * 对比数据记录
     *
     * @param sourceData
     * @param clientData
     * @param checkDataReq
     * @param focusFieldList
     */
    private void compareData(Map<Long, Map<String, Object>> sourceData,
                             Map<Long, Map<String, Object>> clientData,
                             CheckDataReq checkDataReq,
                             List<String> focusFieldList) throws Exception {
        for (Long key : sourceData.keySet()) {
            Map<String, Object> source = sourceData.get(key);
            Map<String, Object> target = clientData.get(key);
            if (target == null) {
                String msg = String.format("client端缺失数据:%s,%s,%ld", checkDataReq.getClientDbName(), checkDataReq.getClientTable(), key);
                log.info(msg);
            }

            for (String fieldName : target.keySet()) {
                if (ignoreFieldList.contains(fieldName)) {
                    continue;
                }
                if (!focusFieldList.contains(fieldName)) {
                    continue;
                }

                if (ObjectUtil.isEmpty(target.get(fieldName)) && ObjectUtil.isEmpty(source.get(fieldName))) {
                    continue;
                }

                if (!ObjectUtil.equal(target.get(fieldName), source.get(fieldName))) {
                    String msg = String.format("数据记录对比不一致:%s,%s,%s,%s", checkDataReq.getClientDbName(), checkDataReq.getClientTable(), key, fieldName);
                    log.info(msg);
                    throw new Exception(msg);
                }
            }
        }
    }

    /**
     * 查询数据量
     *
     * @param checkDataReq
     * @param tableName
     * @param jdbcTemplate
     * @return
     */
    private Integer queryDataCount(CheckDataReq checkDataReq, String tableName, JdbcTemplate jdbcTemplate) {
        String queryCountSql = "select count(*) from " + tableName + " where updated_at >= '" + checkDataReq.getStartTime() + "' and updated_at <= '" + checkDataReq.getEndTime() + "'";

        if (!StringUtils.isEmpty(checkDataReq.getUpdateTimeFieldName())) {
            queryCountSql = "select count(*) from " + tableName + " where " + checkDataReq.getUpdateTimeFieldName() + " >= '" + checkDataReq.getStartTime() + "' and " + checkDataReq.getUpdateTimeFieldName() + " <= '" + checkDataReq.getEndTime() + "'";
        }
        log.info("queryCountSql:{}", queryCountSql);

        Integer totalRecord = jdbcTemplate.queryForObject(queryCountSql, Integer.class);
        return totalRecord;
    }


    /**
     * 查询数据
     *
     * @param checkDataReq
     * @param tableName
     * @param jdbcTemplate
     * @param pageSize
     * @param pageNum
     * @return
     */
    private Map<Long, Map<String, Object>> queryData(CheckDataReq checkDataReq, String tableName, JdbcTemplate jdbcTemplate, Integer pageSize, Integer pageNum) {
        String queryDataSql = "select * from " + tableName + " where updated_at >= '" + checkDataReq.getStartTime() + "' and updated_at <= '" + checkDataReq.getEndTime() + "' order by id asc limit " + (pageNum - 1) * pageSize + "," + pageSize;
        if (!StringUtils.isEmpty(checkDataReq.getUpdateTimeFieldName())) {
            queryDataSql = "select * from " + tableName + " where " + checkDataReq.getUpdateTimeFieldName() + " >= '" + checkDataReq.getStartTime() + "' and " + checkDataReq.getUpdateTimeFieldName() + " <= '" + checkDataReq.getEndTime() + "' order by id asc limit " + (pageNum - 1) * pageSize + "," + pageSize;
        }
        log.info("queryDataSql:{}", queryDataSql);
        log.info("pageNum:{}, pageSize:{}", pageNum, pageSize);

        List<Map<String, Object>> mapList = jdbcTemplate.queryForList(queryDataSql);
        Map<Long, Map<String, Object>> map = mapList.stream().collect(Collectors.toMap(obj -> Long.valueOf(obj.get("id").toString()), obj -> obj));
        log.info("{}", map.size());
        return map;
    }


    /**
     * 构造 dataSource
     *
     * @param cfgDb
     * @return
     */
    private JdbcTemplate buildSourceDataSource(CfgDbDto cfgDb) {
        if (sourceDataSourceMap.containsKey(cfgDb.getId())) {
            return sourceDataSourceMap.get(cfgDb.getId());
        }
        DruidDataSource dataSource = DruidDataSourceBuilder.create().build();
        dataSource.setDriverClassName("com.mysql.jdbc.Driver");
        String url = "jdbc:mysql://" + cfgDb.getDbHost() + ":" + cfgDb.getDbPort() + "/" + cfgDb.getDbName() + "?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true";

        dataSource.setUrl(url);
        dataSource.setUsername(cfgDb.getDbUser());
        dataSource.setPassword(cfgDb.getDbPwd());
        dataSource.setValidationQuery("SELECT 1");//用来检测连接是否有效
        dataSource.setTestOnBorrow(false);
        dataSource.setTestOnReturn(false);
        dataSource.setTestWhileIdle(true);
        dataSource.setTimeBetweenEvictionRunsMillis(600000);
        dataSource.setMaxActive(20);
        dataSource.setInitialSize(10);

        JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
        sourceDataSourceMap.put(cfgDb.getId(), jdbcTemplate);

        return jdbcTemplate;
    }

    /**
     * 构造 dataSource
     *
     * @param cfgClientDb
     * @return
     */
    private JdbcTemplate buildClientDataSource(CfgClientDbEntity cfgClientDb) {
        if (clientDataSourceMap.containsKey(cfgClientDb.getId())) {
            return clientDataSourceMap.get(cfgClientDb.getId());
        }
        DruidDataSource dataSource = DruidDataSourceBuilder.create().build();
        dataSource.setDriverClassName("com.mysql.jdbc.Driver");
        String conUrl = "jdbc:mysql://" + cfgClientDb.getDbHost() + ":" + cfgClientDb.getDbPort() + "/" + cfgClientDb.getClientDbName() + "?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true";

        dataSource.setUrl(conUrl);
        dataSource.setUsername(cfgClientDb.getDbUser());
        dataSource.setPassword(cfgClientDb.getDbPwd());
        dataSource.setValidationQuery("SELECT 1");//用来检测连接是否有效
        dataSource.setTestOnBorrow(false);
        dataSource.setTestOnReturn(false);
        dataSource.setTestWhileIdle(true);
        dataSource.setTimeBetweenEvictionRunsMillis(600000);
        dataSource.setMaxActive(20);
        dataSource.setInitialSize(10);

        JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
        clientDataSourceMap.put(cfgClientDb.getId(), jdbcTemplate);

        return jdbcTemplate;
    }
}
